package com.link.gateway.core.filter.global;

import com.link.gateway.base.user.model.User;
import com.link.gateway.core.cllog.LCLogger;
import com.link.gateway.core.comproperty.*;
import com.link.gateway.core.utils.CacheUtil;
import com.link.gateway.core.utils.RecordInterfaceLogThread;
import com.link.gateway.core.utils.RecordInterfaceLogUtils;
import com.link.gateway.core.utils.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.reactivestreams.Publisher;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.support.BodyInserterContext;
import org.springframework.cloud.gateway.support.CachedBodyOutputMessage;
import org.springframework.cloud.gateway.support.DefaultClientResponse;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ReactiveHttpOutputMessage;
import org.springframework.http.ResponseCookie;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.BodyInserter;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.ExchangeStrategies;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;

import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR;


/**
 * @author guoshan
 * @date 2019年08月29日
 * 判断是否需要记录响应体和请求体
 * 将url接口调用信息通过Kafka传到消息队列
 */
@Component
public class SaveInterfaceLogGlobalFilter implements GlobalFilter, Ordered {

    private static final Logger logger = LogManager.getLogger(SaveInterfaceLogGlobalFilter.class);

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        //跳过检测
        Object ignoreGlobalFilter = exchange.getAttribute(OtherConstants.ATTRIBUTE_IGNORE_GLOBAL_FILTER);
        if (ignoreGlobalFilter != null && (boolean)ignoreGlobalFilter) {
            return chain.filter(exchange);
        }
        URI uri = exchange.getRequest().getURI();
        String origUrl = uri.toString();
        String path = uri.getPath();
        String queryParams = uri.getQuery();
        if (StringUtils.isNotBlank(queryParams)) {
            path =  path + "?" + queryParams ;
        }

        exchange.getAttributes().put(OtherConstants.ORIG_URL, origUrl);
        exchange.getAttributes().put(OtherConstants.REQ_PATH, path);
//        判断是否需要记录接口访问日志
        try {
            String recordLogFlag = CacheUtil.getCorpPropertyValue(OtherConstants.RECORD_INTERFACE_LOG_FLAG);
            if ("Y".equalsIgnoreCase(recordLogFlag)) {
                return saveRespData(exchange,chain);
            }
        } catch (ExecutionException e) {
            LCLogger.withException(e.getMessage());
        }

        return  chain.filter(exchange);
    }

    private Mono<Void> saveRespData(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpResponseDecorator responseDecorator = new ServerHttpResponseDecorator(exchange.getResponse()) {

            @Override
            public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                HttpHeaders headers2 = exchange.getResponse().getHeaders();
                //判断是否需要记录响应体，
                /*
                 String recordRespDataFlag = exchange.getAttribute(ReqAndRespConstants.RECORD_RESPONSE_BODY_FLAG);
                if ("N".equalsIgnoreCase(recordRespDataFlag) || StringUtils.isBlank(recordRespDataFlag)) {
                    recordReqRespMsg(exchange,"");
                    return super.writeWith(body);
                }
                */
                String originalResponseContentType = exchange.getAttribute(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR);
                HttpHeaders httpHeaders = new HttpHeaders();
                httpHeaders.add(HttpHeaders.CONTENT_TYPE, originalResponseContentType);
                SaveInterfaceLogGlobalFilter.ResponseAdapter responseAdapter = new SaveInterfaceLogGlobalFilter.ResponseAdapter(body, httpHeaders);
                DefaultClientResponse clientResponse = new DefaultClientResponse(responseAdapter, ExchangeStrategies.withDefaults());
                Mono<String> rawBody = clientResponse.bodyToMono(String.class).map(s -> s);
                BodyInserter<Mono<String>, ReactiveHttpOutputMessage> bodyInserter = BodyInserters.fromPublisher(rawBody, String.class);
                CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, exchange.getResponse().getHeaders());
                return bodyInserter.insert(outputMessage, new BodyInserterContext())
                        .then(Mono.defer(() -> {
                            Flux<DataBuffer> messageBody = outputMessage.getBody();
                            Flux<DataBuffer> flux = messageBody.map(buffer -> {
                                CharBuffer charBuffer = StandardCharsets.UTF_8.decode(buffer.asByteBuffer());
                                DataBufferUtils.release(buffer);
                                String respBody = charBuffer.toString();
                                recordReqRespMsg(exchange,respBody);
                                return getDelegate().bufferFactory().wrap(respBody.getBytes(StandardCharsets.UTF_8));
                            });
                            HttpHeaders headers = getHeaders();
                            if (!headers.containsKey(HttpHeaders.TRANSFER_ENCODING)) {
                                flux = flux.doOnNext(data -> headers.setContentLength(data.readableByteCount()));
                            }
                            return getDelegate().writeWith(flux);
                        }));
            }
        };
        Long startTime = System.currentTimeMillis();
        /*User currentUser = RunProperties.TEMP_USER.get();
        String tokenScope = RunProperties.SCOPE.get();
        if (currentUser != null) {
            exchange.getAttributes().put("currentUser", currentUser);
        }
        if (tokenScope != null) {
            exchange.getAttributes().put("tokenScope", tokenScope);
        }*/
        exchange.getAttributes().put(ReqAndRespConstants.REQUEST_START_TIME_KEY, startTime);
        return chain.filter(exchange.mutate().response(responseDecorator).build());
    }

    private void recordReqRespMsg(ServerWebExchange exchange, String respBody) {
        User currentUser = exchange.getAttribute(OtherConstants.CURRENT_USER);
        String tokenScope = exchange.getAttribute(OtherConstants.TOKEN_SCOPE);
        String visitType = exchange.getAttribute(InterfaceLogConstants.VISIT_TYPE);
        visitType = visitType == null ? InterfaceLogConstants.NORMAL : visitType;
        RecordInterfaceLogThread interfaceLog =
                new RecordInterfaceLogThread(exchange.getRequest(),visitType,tokenScope,currentUser,exchange,respBody);
        RecordInterfaceLogUtils.recordLoggedInterfaceLog(interfaceLog);

    }

    private class ResponseAdapter implements ClientHttpResponse {

        private final Flux<DataBuffer> flux;
        private final HttpHeaders headers;

        @SuppressWarnings("unchecked")
        private ResponseAdapter(Publisher<? extends DataBuffer> body, HttpHeaders headers) {
            this.headers = headers;
            if (body instanceof Flux) {
                flux = (Flux) body;
            } else {
                flux = ((Mono) body).flux();
            }
        }

        @Override
        public Flux<DataBuffer> getBody() {
            return flux;
        }

        @Override
        public HttpHeaders getHeaders() {
            return headers;
        }

        @Override
        public HttpStatus getStatusCode() {
            return null;
        }

        @Override
        public int getRawStatusCode() {
            return 0;
        }

        @Override
        public MultiValueMap<String, ResponseCookie> getCookies() {
            return null;
        }
    }

    @Override
    public int getOrder() {
        return FilterOrderConstants.SAVE_RESP_BODY;
    }


}
